// Copyright (c) 2020-present,  INSPUR Co, Ltd.  All rights reserved.
// This source code is licensed under Apache 2.0 License.
#include <util/util.h>
#include <db/merge_helper.h>
#include "memory_arena.h"
#include "rocksdb/merge_operator.h"

namespace rocksdb{

//liliupeng 内存存储区
MemArena::MemArena(const InternalKeyComparator& mkeyCmp, const ImmutableCFOptions& ioptions)
    : IMemoryArena(mkeyCmp, ioptions),
      mkey_cmp_(mkeyCmp),
      moptions_(ioptions), refs_(0) {
  int64_t current_time = 0;
  oldest_key_time_ = static_cast<uint64_t >(current_time);
}

MemArena::~MemArena() {
  auto iter = kv_list_.GetIterator();
  iter->SeekToFirst();
  while(iter->Valid()){
    delete[] iter->key();
    iter->Next();
  }
  auto del_iter = range_del_list_.GetIterator();
  del_iter->SeekToFirst();
  while(del_iter->Valid()){
    delete[] del_iter->key();
    del_iter->Next();
  }
}

//key为internal_key,即包含userkey,seq,type
bool MemArena::RangeDeletionInsert(const Slice& key, const Slice& value){
  size_t key_size = key.size();
  size_t val_size = value.size();
  const size_t encoded_len = VarintLength(key_size) +
                               key_size + VarintLength(val_size) +
                               val_size;
  char* buf = new char[encoded_len];
  char* p = EncodeVarint32(buf, key_size);
  memcpy(p, key.data(), key_size);
  p += key_size;
  p = EncodeVarint32(p, val_size);
  memcpy(p, value.data(), val_size);

  auto node = new VersionNode();
  node->SetNext(nullptr);
  node->SetPrev(nullptr);
  bool ok = node->CASSetKey(nullptr, buf);
  assert(ok);

  bool res = range_del_list_.Insert(reinterpret_cast<const char *>(node));
  if(!res){
    delete[] buf;
  }
  return res;
}

//key为internal_key,即包含userkey,seq,type
bool MemArena::KVDataInsert(const Slice& key, const Slice& value){  ////key  value new两个空间   以后再改
  size_t key_size = key.size();
  size_t val_size = value.size();
  const uint32_t encoded_len = VarintLength(key_size) +
                               key_size + VarintLength(val_size) +
                               val_size;
  char* buf = new char[encoded_len];
  char* p = EncodeVarint32(buf, key_size);
  memcpy(p, key.data(), key_size);
  p += key_size;
  p = EncodeVarint32(p, val_size);
  memcpy(p, value.data(), val_size);

  auto node = new VersionNode();
  node->SetNext(nullptr);
  node->SetPrev(nullptr);
  bool ok = node->CASSetKey(nullptr, buf);
  assert(ok);

  bool res = kv_list_.Insert(reinterpret_cast<const char *>(node));
  if(res){
    kv_insert_num_.fetch_add(encoded_len);
  } else{
    delete[] buf;
  }

  return res;
}


//key为internal_key,即包含userkey,seq,type
bool MemArena::Insert(const Slice& key, const Slice& value) {  ////key  value new两个空间   以后再改
  size_t key_size = key.size();
  size_t val_size = value.size();
  const uint32_t encoded_len = VarintLength(key_size) +
                               key_size + VarintLength(val_size) +
                               val_size;
  char* buf = new char[encoded_len];
  char* p = EncodeVarint32(buf, key_size);
  memcpy(p, key.data(), key_size);
  p += key_size;
  p = EncodeVarint32(p, val_size);
  memcpy(p, value.data(), val_size);

  auto node = new VersionNode();
  node->SetNext(nullptr);
  node->SetPrev(nullptr);
  bool ok = node->CASSetKey(nullptr, buf);
  assert(ok);

  bool res = kv_list_.Insert(reinterpret_cast<const char *>(node));
  if(res){
    kv_insert_num_.fetch_add(encoded_len);
  } else{
    delete[] buf;
  }

  return res;
}

//需要判断返回的是不是nullptr,如果是，说明当前key在范围删除内
//key为memtable_key
ArtTree::Iterator* MemArena::Seek(const Slice& key){
  auto iter = this->RangeDeletionSeek(key);
  if(iter->Valid()){
    Slice start = GetLengthPrefixedSlice(iter->key());
    Slice end = GetLengthPrefixedSlice(start.data() + start.size());
    if(key.compare(start) >= 0 && key.compare(end) < 0)  //范围删除左闭右开？？
      return nullptr;      //////改,不是返回范围删除的迭代器
  }
   iter = kv_list_.GetIterator();
   iter->HashSeek(key);
  return iter;
}

ArtTree::Iterator* MemArena::RangeDeletionSeek(const Slice& key){
  auto iter = range_del_list_.GetIterator();
  iter->HashSeek(key);
  return iter;
}

void MemArena::Ref() {
    ++refs_;
}

bool MemArena::Unref() {
    assert(refs_ >= 1);
    --refs_;
    if (refs_ == 0) {
        delete this;
        return true;
    }
    return false;
}

int MemArena::KeyComparator::operator()(const char *prefix_len_key1, const char *prefix_len_key2) const {
  // Internal keys are encoded as length-prefixed strings.
  Slice k1 = GetLengthPrefixedSlice(prefix_len_key1);
  Slice k2 = GetLengthPrefixedSlice(prefix_len_key2);
  return comparator.CompareKeySeq(k1, k2);
}

int MemArena::KeyComparator::operator()(const char *prefix_len_key,
                                        const MemTableRep::KeyComparator::DecodedType &key) const {
  // Internal keys are encoded as length-prefixed strings.
  Slice a = GetLengthPrefixedSlice(prefix_len_key);
  return comparator.CompareKeySeq(a, key);
}

// Callback from MemTable::Get()
namespace {

struct Saver {
    Status* status;
    const LookupKey* key;
    bool* found_final_value;  // Is value set correctly? Used by KeyMayExist
    bool* merge_in_progress;
    std::string* value;
    SequenceNumber seq;
    const MergeOperator* merge_operator;
    // the merge operations encountered;
    MergeContext* merge_context;
    SequenceNumber max_covering_tombstone_seq;
    MemArena* mem_arena;
    Logger* logger;
    Statistics* statistics;
    bool inplace_update_support;
    Env* env_;
    ReadCallback* callback_;
    bool* is_blob_index;

    bool CheckCallback(SequenceNumber _seq) {
        if (callback_) {
            return callback_->IsVisible(_seq);
        }
        return true;
    }
};
}  // namespace

static bool SaveValue(void* arg, const char* entry) {   /////加上锁
    Saver* s = reinterpret_cast<Saver*>(arg);
    assert(s != nullptr);
    MergeContext* merge_context = s->merge_context;
    SequenceNumber max_covering_tombstone_seq = s->max_covering_tombstone_seq;
    const MergeOperator* merge_operator = s->merge_operator;

    assert(merge_context != nullptr);

    // entry format is:
    //    klength  varint32
    //    userkey  char[klength-8]
    //    tag      uint64
    //    vlength  varint32
    //    value    char[vlength]
    // Check that it belongs to same user key.  We do not check the
    // sequence number since the Seek() call above should have skipped
    // all entries with overly large sequence numbers.
    uint32_t key_length;
    const char* key_ptr = GetVarint32Ptr(entry, entry + 5, &key_length);
    if (s->mem_arena->GetInternalKeyComparator().user_comparator()->Equal(
            Slice(key_ptr, key_length - 8), s->key->user_key())) {
        // Correct user key
        const uint64_t tag = DecodeFixed64(key_ptr + key_length - 8);
        ValueType type;
        SequenceNumber seq;
        UnPackSequenceAndType(tag, &seq, &type);
        // If the value is not in the snapshot, skip it
        if (!s->CheckCallback(seq)) {
            return true;  // to continue to the next seq
        }

        s->seq = seq;

        if ((type == kTypeValue || type == kTypeMerge || type == kTypeBlobIndex) &&
            max_covering_tombstone_seq > seq) {
            type = kTypeRangeDeletion;
        }
        switch (type) {
            case kTypeBlobIndex:
                if (s->is_blob_index == nullptr) {
                    ROCKS_LOG_ERROR(s->logger, "Encounter unexpected blob index.");
                    *(s->status) = Status::NotSupported(
                            "Encounter unsupported blob value. Please open DB with "
                            "rocksdb::blob_db::BlobDB instead.");
                } else if (*(s->merge_in_progress)) {
                    *(s->status) =
                            Status::NotSupported("Blob DB does not support merge operator.");
                }
                if (!s->status->ok()) {
                    *(s->found_final_value) = true;
                    return false;
                }
                FALLTHROUGH_INTENDED;
            case kTypeValue: {
                Slice v = GetLengthPrefixedSlice(key_ptr + key_length);
                *(s->status) = Status::OK();
                if (*(s->merge_in_progress)) {
                    if (s->value != nullptr) {
                        *(s->status) = MergeHelper::TimedFullMerge(
                                merge_operator, s->key->user_key(), &v,
                                merge_context->GetOperands(), s->value, s->logger,
                                s->statistics, s->env_, nullptr /* result_operand */, true);
                    }
                } else if (s->value != nullptr) {
                    s->value->assign(v.data(), v.size());
                }
                *(s->found_final_value) = true;
                if (s->is_blob_index != nullptr) {
                    *(s->is_blob_index) = (type == kTypeBlobIndex);
                }
                return false;
            }
            case kTypeDeletion:
            case kTypeSingleDeletion:
            case kTypeRangeDeletion: {
                if (*(s->merge_in_progress)) {
                    if (s->value != nullptr) {
                        *(s->status) = MergeHelper::TimedFullMerge(
                                merge_operator, s->key->user_key(), nullptr,
                                merge_context->GetOperands(), s->value, s->logger,
                                s->statistics, s->env_, nullptr /* result_operand */, true);
                    }
                } else {
                    *(s->status) = Status::NotFound();
                }
                *(s->found_final_value) = true;
                return false;
            }
            case kTypeMerge: {
                if (!merge_operator) {
                    *(s->status) = Status::InvalidArgument(
                            "merge_operator is not properly initialized.");
                    // Normally we continue the loop (return true) when we see a merge
                    // operand.  But in case of an error, we should stop the loop
                    // immediately and pretend we have found the value to stop further
                    // seek.  Otherwise, the later call will override this error status.
                    *(s->found_final_value) = true;
                    return false;
                }
                Slice v = GetLengthPrefixedSlice(key_ptr + key_length);
                *(s->merge_in_progress) = true;
                merge_context->PushOperand(
                        v, false /* operand_pinned */);
                if (merge_operator->ShouldMerge(
                        merge_context->GetOperandsDirectionBackward())) {
                    *(s->status) = MergeHelper::TimedFullMerge(
                            merge_operator, s->key->user_key(), nullptr,
                            merge_context->GetOperands(), s->value, s->logger, s->statistics,
                            s->env_, nullptr /* result_operand */, true);
                    *(s->found_final_value) = true;
                    return false;
                }
                return true;
            }
            default:
                assert(false);
                return true;
        }
    }

    // s->state could be Corrupt, merge or notfound
    return false;
}

bool MemArena::Get(const LookupKey& key, std::string* value, Status* s,
                   MergeContext* merge_context,
                   SequenceNumber* max_covering_tombstone_seq,
                   SequenceNumber* seq, const ReadOptions& read_opts,
                   ReadCallback* callback, bool* is_blob_index) {
    // The sequence number is updated synchronously in version_set.h
    if (GetKVInsertNum()) {
        // Avoiding recording stats for speed.
        return false;
    }
    PERF_TIMER_GUARD(get_from_memtable_time);

    std::unique_ptr<FragmentedRangeTombstoneIterator> range_del_iter(
            NewRangeTombstoneIterator(*seq,
                                      this->mkey_cmp_.comparator));
    if (range_del_iter != nullptr) {
        *max_covering_tombstone_seq =
                std::max(*max_covering_tombstone_seq,
                         range_del_iter->MaxCoveringTombstoneSeqnum(key.user_key()));
    }

    bool found_final_value = false;
    bool merge_in_progress = s->IsMergeInProgress();
    Saver saver{};
    saver.status = s;
    saver.found_final_value = &found_final_value;
    saver.merge_in_progress = &merge_in_progress;
    saver.key = &key;
    saver.value = value;
    saver.seq = kMaxSequenceNumber;
    saver.mem_arena = this;
    saver.merge_context = merge_context;
    saver.max_covering_tombstone_seq = *max_covering_tombstone_seq;
    saver.merge_operator = moptions_.merge_operator;
    saver.logger = moptions_.info_log;
    saver.statistics = moptions_.statistics;
    saver.env_ = moptions_.env;
    saver.callback_ = callback;
    saver.is_blob_index = is_blob_index;
    Get(key, &saver, SaveValue);

    *seq = saver.seq;

    // No change to value, since we have not yet found a Put/Delete
    if (!found_final_value && merge_in_progress) {
        *s = Status::MergeInProgress();
    }
    PERF_COUNTER_ADD(get_from_memtable_count, 1);
    return found_final_value;
}

}
